package com.steakliu.sun.client.netty;

import com.steakliu.sun.buffer.api.DataBuffer;
import com.steakliu.sun.client.api.ClientConfig;
import com.steakliu.sun.client.api.Client;
import com.steakliu.sun.client.netty.manager.ClientHandlerCacheManager;
import com.steakliu.sun.client.netty.handler.SunClientHandler;
import com.steakliu.sun.client.netty.initializer.SunClientChannelInitializer;
import com.steakliu.sun.common.helper.SunServiceHelper;
import com.steakliu.sun.common.meta.ServiceMeta;
import com.steakliu.sun.common.thread.ClientConcurrentThreadPool;
import com.steakliu.sun.protocol.SunProtocol;
import com.steakliu.sun.protocol.SunRequest;
import com.steakliu.sun.protocol.context.RpcContext;
import com.steakliu.sun.protocol.future.SunFuture;
import com.steakliu.sun.registry.api.RegistryService;
import com.steakliu.sun.spi.loader.ExtensionLoader;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 功能说明：
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  01:09
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class NettyClient implements Client {

    private final Logger LOGGER = LoggerFactory.getLogger(NettyClient.class);

    private final Bootstrap bootstrap;
    private final NioEventLoopGroup eventLoopGroup;
    private final ClientConfig config;
    private static volatile NettyClient nettyClient;
    private final RegistryService registryService;
    private final ClientConcurrentThreadPool clientConcurrentThreadPool;
    private  DataBuffer dataBuffer;

    public static NettyClient getInstance(ClientConfig config, RegistryService registryService, ClientConcurrentThreadPool clientConcurrentThreadPool) {
        if (null == nettyClient) {
            synchronized (NettyClient.class) {
                if (null == nettyClient) {
                    nettyClient = new NettyClient(config, registryService, clientConcurrentThreadPool);
                }
            }
        }
        return nettyClient;
    }

    public NettyClient(ClientConfig config, RegistryService registryService, ClientConcurrentThreadPool clientConcurrentThreadPool) {
        this.config = config;
        this.clientConcurrentThreadPool = clientConcurrentThreadPool;
        this.registryService = registryService;
        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
        if (config.isEnableBuffer()){
            this.dataBuffer = ExtensionLoader.getExtension(DataBuffer.class, config.getBufferType());
            this.dataBuffer.init(10000);
        }
        this.buildNettyGroup();
    }

    private void buildNettyGroup() {
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new SunClientChannelInitializer(config, clientConcurrentThreadPool,dataBuffer));
    }


    @Override
    public SunFuture send(SunProtocol<SunRequest> protocol) throws Exception {
        //从本地缓存或者注册中心获取服务元数据
        ServiceMeta serviceMeta = this.getServiceMetaFromLocalCacheOrRegistryCenter(protocol, config.getDirectConnections());
        if (null == serviceMeta) {
            throw new RuntimeException("没有可用服务，请检查服务注册中心连接是否正确或配置是否正确");
        }
        //获取服务对应的handler
        SunClientHandler clientHandler = this.getClientHandler(serviceMeta);
        if (null == clientHandler) {
            throw new RuntimeException("连接不到服务端，请检查服务端是否正常！");
        }
        //发送请求
        SunFuture future = clientHandler.sendRequest(protocol, config);
        if (config.isAsync()){
            RpcContext.setContext(future);
        }else {
            return future;
        }
        return null;
    }

    /**
     * 获取Handler
     *
     * @param serviceMeta
     * @return
     */
    private SunClientHandler getClientHandler(ServiceMeta serviceMeta) {
        String handlerKey = ClientHandlerCacheManager.generateHandlerKey(serviceMeta);
        SunClientHandler clientHandler = ClientHandlerCacheManager.get(handlerKey);
        if (null == clientHandler) {
            clientHandler = this.connectClientHandler(serviceMeta);
            if (null == clientHandler) {
                //重连
                for (int i = 0; i < config.getRetries(); i++) {
                    clientHandler = connectClientHandler(serviceMeta);
                    if (null != clientHandler) {
                        //加入Handler缓存
                        ClientHandlerCacheManager.put(handlerKey, clientHandler);
                        break;
                    }
                }
            } else {
                ClientHandlerCacheManager.put(handlerKey, clientHandler);
            }
        }
        return clientHandler;
    }


    private SunClientHandler connectClientHandler(ServiceMeta serviceMeta) {
        try {
            ChannelFuture future = bootstrap.connect(serviceMeta.getHost(), serviceMeta.getPort()).sync().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        LOGGER.info("Successfully connected to the netty service，host is {}，port is {}", serviceMeta.getHost(), serviceMeta.getPort());
                    } else {
                        LOGGER.info("Failed to connected to the netty service，host is {}，port is {}", serviceMeta.getHost(), serviceMeta.getPort());
                    }
                }
            });
            return future.channel().pipeline().get(SunClientHandler.class);
        } catch (Exception e) {
            LOGGER.info("连接服务端失败！");
            throw new RuntimeException(e);
        }
    }

    private ServiceMeta getServiceMetaFromLocalCacheOrRegistryCenter(SunProtocol<SunRequest> protocol, String directConnections) throws Exception {
        SunRequest request = protocol.getBody();
        String serviceName = SunServiceHelper.getServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        ServiceMeta serviceMeta = registryService.discovery(serviceName, directConnections, config.getLoadBalance());
        if (null == serviceMeta) {
            for (int i = 0; i < config.getRetries(); i++) {
                ServiceMeta meta = registryService.discovery(serviceName, directConnections, config.getLoadBalance());
                if (null != meta) {
                    serviceMeta = meta;
                    break;
                }
            }
        }
        return serviceMeta;
    }

    private void close() {
        eventLoopGroup.shutdownGracefully();
    }
}
